package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.bean.KeyWordStats;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.IkAnalyzer;
import com.atguigu.gmall.realtime.function.KwProduct;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/4/24 9:10
 */
public class ProductKeywordStatsApp extends BaseSQLApp {
    public static void main(String[] args) {
        new ProductKeywordStatsApp().init("ProductKeywordStatsApp", 4004, 1, "ProductKeywordStatsApp");
    }
    
    @Override
    public void handle(StreamTableEnvironment tEnv) {
        
        
        tEnv.executeSql("create table product(" +
                            "   stt string, " +
                            "   edt string," +
                            "   sku_name string, " +
                            "   click_ct bigint, " +
                            "   order_ct bigint," +
                            "   cart_ct bigint " +
                            ")with(" +
                            "  'connector' = 'kafka', " +
                            "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
                            "  'properties.group.id' = 'ProductKeywordStatsApp', " +
                            "  'topic' = '" + Constant.TOPIC_DWS_PRODUCT_STATS + "', " +
                            "  'scan.startup.mode' = 'latest-offset', " +
                            "  'format' = 'json'" +
                            ")");
        
        // 1. 过滤出来三个count中,至少一个不为0的数据
        Table t1 = tEnv.sqlQuery("select " +
                                     "* " +
                                     "from product " +
                                     "where click_ct > 0 " +
                                     "or order_ct > 0 " +
                                     "or cart_ct > 0");
        tEnv.createTemporaryView("t1", t1);
        // 2. 分词
        // 2.1 注册自定义的分词函数
        tEnv.createTemporaryFunction("ik_analyzer", IkAnalyzer.class);
        // 2.2 使用函数实现分词
        Table t2 = tEnv.sqlQuery("select" +
                                     " stt, edt, " +
                                     " word," +
                                     " click_ct, order_ct, cart_ct " +
                                     "from t1 " +
                                     ", lateral table( ik_analyzer(sku_name) ) ");
        tEnv.createTemporaryView("t2", t2);
        // 3. 把三个count列变成3行
        tEnv.createTemporaryFunction("kw_product", KwProduct.class);
        Table t3 = tEnv.sqlQuery("select " +
                                     "stt, " +
                                     "edt," +
                                     "word keyword, " +
                                     "source, " +
                                     "ct " +
                                     "from t2 " +
                                     "join lateral table(kw_product(click_ct, order_ct, cart_ct)) on true");
        tEnv.createTemporaryView("t3", t3);
        
        // 3. 相同窗口关键词和source的进行聚合
        Table result = tEnv.sqlQuery("select " +
                                         "stt, edt, keyword, source, " +
                                         "sum(ct) ct, " +
                                         "unix_timestamp() *1000 ts " +
                                         "from t3 " +
                                         "group by stt, edt, keyword, source");
        
        
        // 3. 写入到CLickhouse中
        tEnv
            .toRetractStream(result, KeyWordStats.class)
            .filter(t -> t.f0)
            .map(t -> t.f1)
            .addSink(FlinkSinkUtil.getClickHouseSink(Constant.CLICKHOUSE_DB, "keyword_stats_2022", KeyWordStats.class));
        
        
    }
}
/*
目前有这样的数据:
窗口     商品       点击量  下单量    加购量
0-5     小米手机   10       5        20
0-5     华为手机   100       50        200
------------------------------------------------
第一步: 分词:
0-5     小米   10       5        20
0-5     手机   10       5        20

0-5     华为   100       50        200
0-5     手机   100       50        200


第二部: 制表
给三个数量, 返回  3行两列


--------------
目的:

窗口    关键词    来源     ct
0-5     小米    "click"   10
0-5     小米    "order"   5
0-5     小米    "cart"    20

0-5     手机    "click"   10
0-5     手机    "order"   5
0-5     手机    "cart"    20

....
0-5     手机    "click"   100
0-5     手机    "order"   50
0-5     手机    "cart"    200
 */